package com.qch.rocket.demo.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.Set;

/**
 * @Author qch
 * @Date 2021/6/8
 */
public class MyConsumer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer();
        //消费模式：集群
        pullConsumer.setMessageModel(MessageModel.CLUSTERING);
        //消费模式：集群
        pullConsumer.setMessageModel(MessageModel.BROADCASTING);
        Set<MessageQueue> messageQueues = pullConsumer.fetchSubscribeMessageQueues("tp_demo_05");
        for (MessageQueue messageQueue : messageQueues) {
            //最后一个参数是单次最大拉取个数
            pullConsumer.pull(messageQueue,"TagA||TagB",0,100);
        }

        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer();
        //消费模式：集群
        pushConsumer.setMessageModel(MessageModel.CLUSTERING);
        //消费模式：集群
        pushConsumer.setMessageModel(MessageModel.BROADCASTING);
        pushConsumer.setConsumeThreadMin(1);
        pushConsumer.setConsumeThreadMax(10);
        //TagA||TagB||TagC
        pushConsumer.subscribe("tp_demo_05","*");
        //设置消息批处理种的一个批次种消息的最大个数
        pushConsumer.setConsumeMessageBatchMaxSize(10);
        //在设置完之后调用start方法初始化并运行推送消息的消费者
        pullConsumer.start();
    }
}
